package com.zengfa.task.client.service.impl;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.zengfa.task.api.enums.JobStatus;
import com.zengfa.task.api.enums.RunType;
import com.zengfa.task.api.model.JobDTO;
import com.zengfa.task.api.model.JobNodeDTO;
import com.zengfa.task.api.utils.JobUtil;
import com.zengfa.task.client.config.TaskNodeConfig;
import com.zengfa.task.client.model.TaskDefine;
import com.zengfa.task.client.model.FlowNodeDefine;
import com.zengfa.task.client.model.FlowNodeMonitorDefine;
import com.zengfa.task.client.model.TaskCallbackDefine;
import com.zengfa.task.client.service.JobWorkerService;
import com.zengfa.task.client.support.ServerApiSupport;
import com.zengfa.task.client.vo.JobContextVo;
import com.zengfa.task.client.vo.JobResult;

import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class JobWorkerServiceImpl implements JobWorkerService {

	@Autowired
	private ServerApiSupport serverApiSupport;
	@Autowired
	private ApplicationContext applicationContext;
	@Autowired
	private Environment environment;
	
	private final Map<String, JobRunnable> runningJobs = new ConcurrentHashMap<>();
	
	@Override
	public void runFlowTask(JobDTO jobDTO) {
		//判断任务状态，如果不是NEW状态，则重置任务状态
		JobStatus jobStatus = jobDTO.getStatus();
		if(JobStatus.WAIT != jobStatus) {
			serverApiSupport.resetJob(jobDTO.getJobId());
		}
		JobNodeDTO startNode = serverApiSupport.getNextRunNode(jobDTO.getJobId());
		log.info("获取启动节点" + JSONObject.toJSONString(startNode));
		runFlowTask(startNode);
	}
	
	private void runFlowTask(JobNodeDTO startNode) {
		JobContextVo jobContextVo = initJobContext(startNode);
		jobContextVo.setCurrentNode(startNode);
		JobDTO jobDTO = startNode.getJobDTO();
		JobRunnable jobRunnable = jobNodeAction(jobContextVo, jobDTO);
		ThreadPoolTaskExecutor executor = JobUtil.getTaskExecutor(jobDTO.getJobGroup());
		CompletableFuture<Void> future = CompletableFuture.runAsync(jobRunnable, executor);
		runningJobs.put(jobDTO.getJobId(), jobRunnable);
//		future.whenComplete((Void,e) -> {
//			dealTaskCallback(jobDTO.getJobId(), jobContextVo);
//		});
	}
	
	private JobContextVo initJobContext(JobNodeDTO jobNodeDTO) {
		JobDTO jobDTO = jobNodeDTO.getJobDTO();
		JobContextVo jobContextVo = new JobContextVo();
		jobContextVo.setBizId(jobDTO.getBizId());
		jobContextVo.setJobId(jobNodeDTO.getJobId());
//		if(operUser == null) {
//			operUser = userService.findByUserName(jobDTO.getCreatedBy());
//		}
//		jobContextVo.setUser(operUser);
		jobContextVo.setApplicationContext(applicationContext);
		return jobContextVo;
	}
	
	private JobRunnable jobNodeAction(JobContextVo jobContextVo,JobDTO jobDTO) {
//		User user = jobContextVo.getUser();
		Runnable runnable = () -> {
			log.info("线程执行任务");
//            MyMetaObjectHandler.USER_THREAD_LOCAL.set(user);
			JobNodeDTO currentNode = jobContextVo.getCurrentNode();
			String port = environment.getProperty("server.port");
			String ip = JobUtil.getLocalIp()+":"+port;
			currentNode.setExecIp(ip);
			try {
				while(currentNode != null && !Thread.currentThread().isInterrupted()) {
					boolean flag = serverApiSupport.saveJobNodeStatus(currentNode,JobStatus.RUNNING,null);
					if(!flag || Thread.currentThread().isInterrupted()) {
						break;
					}
					TaskDefine flowDefine = TaskNodeConfig.TASKS.get(jobDTO.getJobGroup());
					if(flowDefine == null) {
						break;
					}
					FlowNodeDefine defineNode = flowDefine.getDefineNode(currentNode.getNodeCode());
					Method method = defineNode.getMethod();
					Object serevice = defineNode.getService();
					method.invoke(serevice, jobContextVo);
					RunType runType = currentNode.getRunType();
					if(runType != null && runType == RunType.ASYNC) {
						break;
					}
					flag = serverApiSupport.saveJobNodeStatus(currentNode,JobStatus.SUCCESS,null);
					if(!flag || Thread.currentThread().isInterrupted()) {
						break;
					}
					currentNode = serverApiSupport.getNextRunNode(currentNode.getJobId());
					jobContextVo.setCurrentNode(currentNode);
				}
//				JobNodeDTO nextNode = serverApiSupport.getNextRunNode(currentNode.getJobId());
				if (currentNode == null) {
					//没有下个任务节点，任务完成
					runningJobs.remove(jobDTO.getJobId());
					dealTaskCallback(jobDTO.getJobId(), jobContextVo);
				}
			} catch(Exception e) {
				log.error("任务运行失败:" , e);
				serverApiSupport.saveJobNodeStatus(currentNode,JobStatus.ERROR,e.getMessage());
				runningJobs.remove(jobDTO.getJobId());
				dealTaskCallback(jobDTO.getJobId(), jobContextVo);
			} finally {
//				MyMetaObjectHandler.USER_THREAD_LOCAL.remove();
			}
		};
		return new JobRunnable(runnable,jobContextVo);
	}
	
	class JobRunnable implements Runnable {
		private volatile Thread t = null;
//		private volatile boolean stopped = false;
		private Runnable runnable;
		private JobContextVo jobContext;
		public JobRunnable(Runnable runnable) {
			this.runnable = runnable;
		}
		public JobRunnable(Runnable runnable, JobContextVo jobContext) {
			this.runnable = runnable;
			this.jobContext = jobContext;
		}
		@Override
		public void run() {
			t = Thread.currentThread();
			runnable.run();
		}
		public synchronized void stop(){
//			this.stopped = true;
			t.interrupt();
//			if(t != null && t.isAlive()) {
//				t.stop();
//			}
		}
		public boolean isStopped() {
			return t.isInterrupted();
		}
		public JobContextVo getJobContext() {
			return this.jobContext;
		}
	}

//	private static volatile Set<String> syncJobs = new ConcurrentSkipListSet<>();
	
	/*
	@Override
	public void syncJobStatus(String jobId) {
		if(syncJobs.contains(jobId)) {
			return;
		}
		if(!syncJobs.add(jobId)) {
			return;
		}
		JobRunnable _jobRunnable = runningJobs.get(jobId);
		if(_jobRunnable != null) {
			// 还在运行中
			return;
		}
		JobDTO jobDTO = serverApiSupport.getJob(jobId);
		JobType jobType = jobDTO.getJobType();
		if (jobType == JobType.FLOW) {
			syncFlowJobRunningStatus(jobDTO);
		} else if (jobType == JobType.SINGLE) {
			// 运行完毕,以下两种情况都可以尝试将任务状态由运行中改为失败
			// 1.在同步任务前运行完毕
			// 2.在同步任务后，任务还运行中，但同步任务运行到这里时，任务已经完毕(成功或失败或停止)
			serverApiSupport.saveJobStatus(jobDTO,JobStatus.ERROR, null);
		}
		syncJobs.remove(jobId);
	}
	
	private void syncFlowJobRunningStatus(JobDTO jobDTO) {
		JobNodeDTO currentNode = serverApiSupport.getCurrentRunNode(jobDTO.getJobId());
		TaskDefine flowDefine = TaskNodeConfig.TASKS.get(jobDTO.getJobGroup());
		FlowNodeDefine defineNode = flowDefine.getDefineNode(currentNode.getNodeCode());
		FlowNodeMonitorDefine monitor = defineNode.getFlowNodeMonitorDefine();
		if (monitor == null) {
			return;
		}
		JobContextVo jobContextVo = initJobContext(currentNode);
		Method method = monitor.getMethod();
		try {
			JobResult result = (JobResult)method.invoke(monitor.getService(), jobContextVo);
			log.info("同步任务节点状态结果:"+JSONObject.toJSONString(result));
			if(result == null) {
				return;
			}
			if(JobStatus.RUNNING == result.getStatus()) {
				return;
			}
			boolean flag = serverApiSupport.saveJobNodeStatus(currentNode,result.getStatus(),result.getMessage());
			if(!flag) {
				return;
			}
			JobNodeDTO nextNode = serverApiSupport.getNextRunNode(jobDTO.getJobId());
			if (nextNode != null) {
				jobContextVo.setCurrentNode(nextNode);
				JobRunnable jobRunnable = jobNodeAction(jobContextVo, jobDTO);
				ThreadPoolTaskExecutor executor = JobUtil.getTaskExecutor(jobDTO.getJobGroup());
				CompletableFuture<Void> future = CompletableFuture.runAsync(jobRunnable, executor);
				runningJobs.put(jobDTO.getJobId(), jobRunnable);
			} else {
				dealTaskCallback(jobDTO.getJobId(), jobContextVo);
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	*/

	@Override
	public void runTask(JobDTO jobDTO, Runnable runnable) {
		String port = environment.getProperty("server.port");
		String ip = JobUtil.getLocalIp()+":"+port;
		jobDTO.setExecIp(ip);
		boolean flag = serverApiSupport.saveJobStatus(jobDTO,JobStatus.RUNNING, null);
		if(!flag) {
			return;
		}
		JobRunnable jobRunnable = new JobRunnable(runnable);
		CompletableFuture<Void> future = CompletableFuture.runAsync(() ->{
			jobRunnable.run();
		});
		runningJobs.put(jobDTO.getJobId(), jobRunnable);
		try {
			future.get();
			serverApiSupport.saveJobStatus(jobDTO, JobStatus.RUNNING, null);
		} catch (Exception e1) {
			String errorMsg = e1.getMessage();
			serverApiSupport.saveJobStatus(jobDTO, JobStatus.ERROR, errorMsg);
			throw new RuntimeException(e1);
		}finally {
			runningJobs.remove(jobDTO.getJobId());
		}
	}

	@Override
	public void runAsyncTask(JobDTO jobDTO, Runnable runnable) {
		String port = environment.getProperty("server.port");
		String ip = JobUtil.getLocalIp()+":"+port;
		jobDTO.setExecIp(ip);
		boolean flag = serverApiSupport.saveJobStatus(jobDTO, JobStatus.RUNNING, null);
		if(!flag) {
			return;
		}
		ThreadPoolTaskExecutor executor = JobUtil.getTaskExecutor(jobDTO.getJobGroup());
		JobRunnable jobRunnable = new JobRunnable(runnable);
		CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
			jobRunnable.run();
		}, executor);
		runningJobs.put(jobDTO.getJobId(), jobRunnable);
		future.whenComplete((Void,e) -> {
			if(e != null) {
				log.error("捕获任务异常", e);
				String errorMsg = e.getMessage();
				serverApiSupport.saveJobStatus(jobDTO, JobStatus.ERROR, errorMsg);
			}else {
				serverApiSupport.saveJobStatus(jobDTO, JobStatus.SUCCESS, null);
			}
			dealTaskCallback(jobDTO.getJobId(), null);
			runningJobs.remove(jobDTO.getJobId());
		});
	}

	/*
	@Override
	public void stopTask(String jobId, String msg) {
		syncJobs.add(jobId);
		JobNodeDTO currentNode = serverApiSupport.getCurrentRunNode(jobId);
		if(!StringUtils.hasText(msg)) {
			msg = "主动终止";
		}
		if(currentNode != null) {
			serverApiSupport.saveJobNodeStatus(currentNode,JobStatus.STOP,msg);
		}else {
			JobDTO jobDTO = new JobDTO();
			jobDTO.setJobId(jobId);
			serverApiSupport.saveJobStatus(jobDTO,JobStatus.STOP,msg);
		}
		dealTaskCallback(jobId, null);
		JobRunnable _jobRunnable = runningJobs.remove(jobId);
		if(_jobRunnable != null) {
			_jobRunnable.stop();
		}
		syncJobs.remove(jobId);
	}
	*/
	
	private void dealTaskCallback(String jobId,JobContextVo jobContextVo) {
		JobDTO jobDTO = serverApiSupport.getJob(jobId);
		TaskDefine taskDefine = TaskNodeConfig.TASKS.get(jobDTO.getJobGroup());
		TaskCallbackDefine callback = taskDefine.getTaskCallbackDefine();
		if (callback == null) {
			return;
		}
		if (jobContextVo == null) {
			jobContextVo = new JobContextVo();
			jobContextVo.setBizId(jobDTO.getBizId());
			jobContextVo.setJobId(jobDTO.getJobId());
			jobContextVo.setApplicationContext(applicationContext);
		}
		JobResult jobResult = new JobResult();
		jobResult.setStatus(jobDTO.getStatus());
		jobResult.setMessage(jobDTO.getMessage());
		Method method = callback.getMethod();
		try {
			method.invoke(callback.getService(), jobContextVo, jobResult);
		} catch (Exception e) {
			e.printStackTrace();
		} 
	}
	
	/**
	 * 同步任务状态
	 */
	@Scheduled(cron="0/1 * * * * ? ")
	public void syncRunningWorkStatus() {
		Set<String> jobIds = runningJobs.keySet();
		jobIds.forEach((jobId) -> {
			JobDTO jobDTO = serverApiSupport.getJob(jobId);
			if (jobDTO.getStatus() != JobStatus.RUNNING) {
				// 手动停止时，服务端任务状态为停止，客户端任务线程需要停止
				JobRunnable _jobRunnable = runningJobs.remove(jobId);
				if (_jobRunnable != null) {
					_jobRunnable.stop();
				}
			} else {
				//同步运行中的任务状态,需要根据异步节点判断
				JobRunnable _jobRunnable = runningJobs.get(jobId);
				JobNodeDTO currentNode = _jobRunnable.getJobContext().getCurrentNode();
				if (currentNode != null && currentNode.getRunType() == RunType.ASYNC) {
					syncJobNode(_jobRunnable.getJobContext(), currentNode);
				}
			}
		});
	}
	
	/**
	 * 心跳检查任务
	 */
	@Scheduled(cron="0/30 * * * * ? ")
	public void syncJobHeartbeat() {
		Set<String> jobIds = runningJobs.keySet();
		jobIds.forEach((jobId) -> {
			serverApiSupport.saveJobHeartbeat(jobId);
		});
	}
	
	private void syncJobNode(JobContextVo jobContext, JobNodeDTO currentNode) {
		JobDTO jobDTO = currentNode.getJobDTO();
		TaskDefine flowDefine = TaskNodeConfig.TASKS.get(jobDTO.getJobGroup());
		FlowNodeDefine defineNode = flowDefine.getDefineNode(currentNode.getNodeCode());
		FlowNodeMonitorDefine monitor = defineNode.getFlowNodeMonitorDefine();
		if (monitor == null) {
			return;
		}
		JobContextVo jobContextVo = initJobContext(currentNode);
		Method method = monitor.getMethod();
		try {
			JobResult result = (JobResult)method.invoke(monitor.getService(), jobContextVo);
			log.info("同步任务节点状态结果:"+JSONObject.toJSONString(result));
			if(result == null) {
				return;
			}
			if(JobStatus.RUNNING == result.getStatus()) {
				return;
			}
			boolean flag = serverApiSupport.saveJobNodeStatus(currentNode,result.getStatus(),result.getMessage());
			if(!flag) {
				return;
			}
			JobNodeDTO nextNode = serverApiSupport.getNextRunNode(jobDTO.getJobId());
			if (nextNode != null) {
				jobContextVo.setCurrentNode(nextNode);
				JobRunnable jobRunnable = jobNodeAction(jobContextVo, jobDTO);
				ThreadPoolTaskExecutor executor = JobUtil.getTaskExecutor(jobDTO.getJobGroup());
				CompletableFuture<Void> future = CompletableFuture.runAsync(jobRunnable, executor);
				runningJobs.put(jobDTO.getJobId(), jobRunnable);
			} else {
				dealTaskCallback(jobDTO.getJobId(), jobContextVo);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
